package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.bean.TrafficIsNewPageViewBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Classname DwsTrafficIsNewPageViewWindow
 * @Description TODO
 * @Date 2023/2/17 19:59
 * @Created by lzx
 */
public class DwsTrafficIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);


        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new HashMapStateBackend());
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/eduRealTime/ck");
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        DataStreamSource<String> pageDS = env.addSource(KafkaUtils.getFlinkKafkaConsumer("dwd_traffic_page_log", "dws_traffic_isNew_page_view_window_group"));
        SingleOutputStreamOperator<TrafficIsNewPageViewBean> process = pageDS.map(JSON::parseObject)
                .keyBy(value -> value.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, TrafficIsNewPageViewBean>() {
                    private ValueState<String> lastDateState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                    /*
                    在open方法中,初始化值状态,设置状态TTL为1天
                     */
                        ValueStateDescriptor<String> lastDateDescriptor = new ValueStateDescriptor<>("last_value", String.class);
                        lastDateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastDateState = getRuntimeContext().getState(lastDateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, TrafficIsNewPageViewBean>.Context ctx, Collector<TrafficIsNewPageViewBean> out) throws Exception {
                        String lastDate = lastDateState.value();
                        Long uuCt = 0L;
                        Long ts = value.getLong("ts");
                        //如果状态中的值为空,或者值不为空,但是与当前数据的访问日期不是同一天,则均说明该条数据是主页的独立访客
                        if (StringUtils.isEmpty(lastDate) || !lastDate.equals(DateFormatUtil.toDate(ts))) {
                            //将记录主界面独立访客的变量改为1
                            uuCt = 1L;
                            //将值状态更新为当前日期
                            lastDateState.update(DateFormatUtil.toDate(ts));
                        }
                        Long pvCt = 1L;
                        Long durSum = value.getJSONObject("page").getLong("during_time");
                        String isNew = value.getJSONObject("common").getString("is_new");
                        out.collect(new TrafficIsNewPageViewBean("", "", isNew, uuCt, pvCt, durSum, ts));
                    }
                });

        //process.print();
        SingleOutputStreamOperator<TrafficIsNewPageViewBean> reduce = process
                .assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficIsNewPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((e,r)->e.getTs()))
                .keyBy(TrafficIsNewPageViewBean::getIsNew)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficIsNewPageViewBean>() {
                    @Override
                    public TrafficIsNewPageViewBean reduce(TrafficIsNewPageViewBean value1, TrafficIsNewPageViewBean value2) throws Exception {
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                        return value1;
                    }
                }, new ProcessWindowFunction<TrafficIsNewPageViewBean, TrafficIsNewPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TrafficIsNewPageViewBean, TrafficIsNewPageViewBean, String, TimeWindow>.Context context, Iterable<TrafficIsNewPageViewBean> elements, Collector<TrafficIsNewPageViewBean> out) throws Exception {
                        TrafficIsNewPageViewBean next = elements.iterator().next();
                        long end = context.window().getEnd();
                        long start = context.window().getStart();
                        next.setStt(DateFormatUtil.toYmdHms(start));
                        next.setEdt(DateFormatUtil.toYmdHms(end));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        //reduce.print(">>");
        reduce.addSink(ClickHouseUtils.getClickHouseSinkFunction("insert into dws_traffic_isNew_page_view_window values(?,?,?,?,?,?,?)"));
        env.execute();
    }
}
